package com.navinfo.platform.hive2mysql.service;

import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.common.utils.ConnectionUtil;
import com.navinfo.platform.common.utils.DateUtils;
import com.navinfo.platform.common.utils.StatisticsDef;
import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;
import scala.collection.JavaConverters;
import scala.collection.Seq;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;

public enum ExportHandleService implements IDataHandleService {
    //
    INS;

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        String day = configMap.get(ICommonService.DAY_VALUE);
        int type = configMap.get("args_2") == null ? 0 : Integer.parseInt(configMap.get("args_2"));//0:全部,1:天,2:周,3:月
        Dataset<Row> basicDataset = castValue(obj.get(0));
        Dataset<Row> exDataset = castValue(obj.get(1));
        Dataset<Row> saveDataset = basicDataset.join(exDataset, usingColumn(), "left_outer");
        saveDataset.cache();
        Properties connectionProperties = getDatabaseConfig(configMap);
        String url = configMap.get("mysql.url");
        ConnectionUtil.getInstance().init(configMap);
        Connection conn = ConnectionUtil.getInstance().connect();
        QueryRunner queryRunner = new QueryRunner();
        String tablesPrefix = configMap.get("mysql.tables.prefix");
        String year = day.substring(2, 4);
        if (tablesPrefix == null) {
            return;
        }

        List<String> tableNameList = new ArrayList<>(Arrays.asList(tablesPrefix.split(",")));

        try {

            tableNameList.forEach(tableName -> {
                String deleteSql;
                String date = day;
                switch (type) {
                    case 1:
                        deleteSql = String.format("delete from %s_%s_%s where day = ?", tableName, "day", year);
                        break;
                    case 2:
                        Tuple2<String, String> weekDateTuple = DateUtils.getStartEndDay(day, StatisticsDef.WEEK);
                        date = weekDateTuple._1();
                        deleteSql = String.format("delete from %s_%s_%s where day = ?", tableName, "week", year);
                        break;
                    case 3:
                        Tuple2<String, String> monthDateTuple = DateUtils.getStartEndDay(day, StatisticsDef.MONTH);
                        date = monthDateTuple._1();
                        deleteSql = String.format("delete from %s_%s_%s where day = ?", tableName, "month", year);
                        break;
                    default:
                        deleteSql = String.format("delete from %s_%s_%s where day = ?", tableName, "day", year);
                        date = day;
                }

                try {
                    queryRunner.update(conn, deleteSql, date);
                } catch (SQLException e) {
                    e.printStackTrace();
                }

            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            DbUtils.closeQuietly(conn);
        }

        tableNameList.forEach(tableNamePrefix -> {
            String fieldKey = configMap.get(tableNamePrefix);
            String fields = configMap.get(fieldKey);
            String t;
            switch (type) {
                case 1:
                    t = "day";
                    break;
                case 2:
                    t = "week";
                    break;
                case 3:
                    t = "month";
                    break;
                default:
                    t = "day";
            }

            String tableName = String.format("%s_%s_%s", tableNamePrefix, t, year);
            saveDataset.selectExpr(fields.split(","))
                    .repartition(1)
                    .write()
                    .mode(SaveMode.Append)
                    .jdbc(url, tableName, connectionProperties);
        });
    }

    private Properties getDatabaseConfig(Map<String, String> configMap) {
        String user = configMap.get("mysql.user");
        String password = configMap.get("mysql.password");
        String driver = configMap.get("mysql.driver");
        Properties connectionProperties = new Properties();
        connectionProperties.put("user", user);
        connectionProperties.put("password", password);
        connectionProperties.put("driver", driver);
        return connectionProperties;
    }

    private Seq<String> usingColumn() {
        return JavaConverters.collectionAsScalaIterableConverter(Arrays.asList("tid", "uid", "day")).asScala().toSeq();
    }
}
